package vertx;

import io.vertx.core.AbstractVerticle;
import io.vertx.core.Vertx;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.eventbus.EventBus;
import io.vertx.core.eventbus.MessageConsumer;
import io.vertx.core.net.NetClient;
import io.vertx.core.net.NetClientOptions;
import io.vertx.core.net.NetSocket;
import netty.util.TimeDelay;

import java.util.Arrays;
import java.util.concurrent.CountDownLatch;

public class PushClientNode extends AbstractVerticle {

    String host = System.getProperty("publicAddress", "127.0.0.1");

    EventBus eventBus = null;

    NetClient client = null;

    NetSocket socket;

    static CountDownLatch count = new CountDownLatch(1);

    TimeDelay timeDelay = TimeDelay.getInstance();


    @Override
    public void start() throws Exception {
        eventBus = vertx.eventBus();

        NetClientOptions options = new NetClientOptions();
        options.setTcpNoDelay(true);
        client = vertx.createNetClient(options).connect(1235, host, res -> {
            if (res.succeeded()) {
                System.out.println("Connected!");
                socket = res.result();

                socket.handler(buffer -> {
                    Arrays.stream(buffer.toString().split(",")).forEach(time -> {
                        if (!"".equals(time)) {
                            timeDelay.analyze(Long.parseLong(time));
                        }
                    });
                });

                count.countDown();
            } else {
                System.out.println("Failed to connect: " + res.cause().getMessage());
            }
        });

        MessageConsumer<String> consumer = eventBus.consumer("zetas.address.1");
        consumer.handler(message -> {
            socket.write(Buffer.buffer(message.body()));
        });

    }

    public static void main(String[] args) throws Exception {
        Vertx vertx = Vertx.vertx();
        vertx.deployVerticle(PushClientNode.class.getName());

        count.await();

        while (!Thread.currentThread().isInterrupted()) {
            Thread.sleep(1000);

            StringBuilder sb = new StringBuilder();
            for (int i = 0; i < 100; i++) {
                sb.append(System.nanoTime()).append(",");
            }
            vertx.eventBus().publish("zetas.address.1", sb.toString());
        }
    }
}
